{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "034af843",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 电影数据分析\n",
    "# 1 用户评分平均分\n",
    "# 2 电影平均分\n",
    "# 3 评分大于平均分的电影数量\n",
    "# 4 高分电影（大于3分）打分次数最多的用户\n",
    "# 5 每个用户评分的平均、最高、最低分\n",
    "# 6 查询评分超过100次的电影和平均分和排名top10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "dd3c44c8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import functions as F"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1daf51c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "99b08bfc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+----+---------+\n",
      "|userId|movieId|rate|       ts|\n",
      "+------+-------+----+---------+\n",
      "|     1|   1193|   5|978300760|\n",
      "|     1|    661|   3|978302109|\n",
      "|     1|    914|   3|978301968|\n",
      "|     1|   3408|   4|978300275|\n",
      "|     1|   2355|   5|978824291|\n",
      "+------+-------+----+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#数据结构\n",
    "schema = StructType() \\\n",
    "    .add(\"userId\", StringType())\\\n",
    "    .add(\"movieId\", StringType())\\\n",
    "    .add(\"rate\", IntegerType())\\\n",
    "    .add(\"ts\", StringType())\n",
    "#读数据\n",
    "df = spark.read.format('csv') \\\n",
    "    .option(\"sep\", \"::\") \\\n",
    "    .option(\"header\", False)\\\n",
    "    .option(\"encoding\", \"utf-8\")\\\n",
    "    .schema(schema=schema)\\\n",
    "    .load(\"C:/Users/Administrator/Desktop/demo/ml-1m/ratings.dat\")\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "3c37c77a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+\n",
      "|userId|avg_rate|\n",
      "+------+--------+\n",
      "|   283|    4.96|\n",
      "|  2339|    4.96|\n",
      "|  3324|     4.9|\n",
      "|  3902|    4.89|\n",
      "|   447|    4.84|\n",
      "|   446|    4.84|\n",
      "|  4649|    4.82|\n",
      "|  4634|    4.81|\n",
      "|  1131|     4.8|\n",
      "|  4925|    4.76|\n",
      "|  5069|    4.76|\n",
      "|  4755|    4.76|\n",
      "|   682|    4.73|\n",
      "|  4801|    4.73|\n",
      "|  1670|    4.71|\n",
      "|    91|     4.7|\n",
      "|  3461|     4.7|\n",
      "|  5984|     4.7|\n",
      "|  5768|     4.7|\n",
      "|  2155|    4.69|\n",
      "+------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 1 用户评分平均分\n",
    "df.groupBy(\"userId\")\\\n",
    "    .avg(\"rate\")\\\n",
    "    .withColumnRenamed(\"avg(rate)\", \"avg_rate\")\\\n",
    "    .withColumn('avg_rate', F.round(\"avg_rate\",2))\\\n",
    "    .orderBy(\"avg_rate\", ascending=False)\\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "bbaf9973",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------+\n",
      "|userId|avg_rate|\n",
      "+------+--------+\n",
      "|   283|    4.96|\n",
      "|  2339|    4.96|\n",
      "|  3324|     4.9|\n",
      "|  3902|    4.89|\n",
      "|   447|    4.84|\n",
      "|   446|    4.84|\n",
      "|  4649|    4.82|\n",
      "|  4634|    4.81|\n",
      "|  1131|     4.8|\n",
      "|  4925|    4.76|\n",
      "|  5069|    4.76|\n",
      "|  4755|    4.76|\n",
      "|   682|    4.73|\n",
      "|  4801|    4.73|\n",
      "|  1670|    4.71|\n",
      "|    91|     4.7|\n",
      "|  3461|     4.7|\n",
      "|  5984|     4.7|\n",
      "|  5768|     4.7|\n",
      "|  2155|    4.69|\n",
      "+------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.createOrReplaceTempView(\"ratings\")\n",
    "spark.sql(\n",
    "    \"\"\"\n",
    "        select userId, round(avg(rate),2) as avg_rate \n",
    "        from ratings\n",
    "        group by userId\n",
    "        order by avg_rate desc\n",
    "    \"\"\"\n",
    ").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "c3c90bbc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "575281\n"
     ]
    }
   ],
   "source": [
    "#3 评分大于平均分的电影数量\n",
    "avg = df.select(F.avg(df['rate'])).first()['avg(rate)']\n",
    "result = df.where(df['rate'] > avg).count()\n",
    "print(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "7bcf77ba",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|  575281|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"\n",
    "    select count(*) from ratings\n",
    "    where rate > (select avg(rate) from ratings)\n",
    "\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "85479ae2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|round(avg(rate), 0)|\n",
      "+-------------------+\n",
      "|                4.0|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# 4 高分电影（大于3分）打分次数最多的用户的平均打分\n",
    "uid = df.where(\"rate>3\")\\\n",
    "    .groupBy(\"userId\")\\\n",
    "    .count()\\\n",
    "    .orderBy(\"count\", ascending=False)\\\n",
    "    .limit(1)\\\n",
    "    .first()[\"userId\"]\n",
    "df.where(df['userId']==uid).select(F.round(F.avg(\"rate\"))).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "4f6e1e91",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|round(avg(rate), 0)|\n",
      "+-------------------+\n",
      "|                4.0|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2 = spark.sql(\"\"\"\n",
    "    select userId, count(rate) as cnt\n",
    "    from ratings\n",
    "    where rate>3\n",
    "    group by userId\n",
    "    order by cnt desc\n",
    "    limit 1\n",
    "\"\"\")\n",
    "sql = \"select round(avg(rate)) from ratings where userId='%s'\" % df2.first()['userId']\n",
    "spark.sql(sql).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "caa401e7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|test1(rate)|\n",
      "+-----------+\n",
      "|         10|\n",
      "+-----------+\n",
      "only showing top 1 row\n",
      "\n",
      "+-----------+\n",
      "|test1(rate)|\n",
      "+-----------+\n",
      "|         10|\n",
      "+-----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# UDF\n",
    "# spark.udf.register(\"方法名称\"， 函数， 类型)\n",
    "spark.udf.register(\"test1\", lambda x: x*2, IntegerType())\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "    select test1(rate) from ratings\n",
    "\"\"\").show(1)\n",
    "\n",
    "df.selectExpr(\"test1(rate)\").show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "4e99dc8b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|<lambda>(rate)|\n",
      "+--------------+\n",
      "|            10|\n",
      "|             6|\n",
      "|             6|\n",
      "|             8|\n",
      "|            10|\n",
      "+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# udf1 = F.udf(\"方法\",\"返回值\")\n",
    "udf1 = F.udf(lambda x: x*2, IntegerType())\n",
    "df.select(udf1(df['rate'])).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a54c0dc2",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
